package com.cheney.demo.controller;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.concurrent.TimeUnit;

/**
 * @author cheney
 */
@RestController
@RequestMapping("/kafka")
public class KafkaController {

    /**
     * 注入kafkaTemplate
     */
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    /**
     * 生产者-测试key和partition
     */
    @RequestMapping("/test")
    public String test() {
        int iMax = 6;
        for (int i = 1; i < iMax; i++) {
            // topic 名称 key data 消息数据
            kafkaTemplate.send("my-topic", "key" + i, "data" + i);
        }
        return "success";
    }

    /**
     * 生产者-异步回调
     */
    @RequestMapping("/testAsync")
    public String testAsync(String key) {
        //发送后异步回调，在主线程完毕后执行
        kafkaTemplate.send("my-topic", "key" + key, "data" + key).addCallback(result -> {
            try {
                TimeUnit.SECONDS.sleep(5);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("发送成功==>" + result.toString());
        }, ex -> {
            throw new RuntimeException(ex);
        });
        return "success";
    }

}
